package com.ruirui.demo;

import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.SignalType;

import java.util.function.Consumer;

public class SubscribeDemo {
    public static void main(String[] args) {
        Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5);
        //无参 让流动起来
        flux.subscribe();
        //传消费者
        flux.subscribe(System.out::println);
        //传两个参数 感知异常结束
        flux.subscribe(e->System.out.println("e1"+e/0),
                throwable -> System.out.println("error"+throwable));
        //传三个参数 感知正常结束
        flux.subscribe(e->System.out.println("e1"),
                throwable -> System.out.println("error"+throwable),
                ()->System.out.println("complete"));
        //自定义消费者 推荐从baseSubscriber开干
        flux.subscribe(new BaseSubscriber<Integer>() {
            /**
             * 订阅的时候触发
             * @param subscription the subscription to optionally process
             */
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                System.out.println("绑定了"+subscription);
                //要数据 请求几次数据 每次请求代表总共能得到buffsize个数据
                request(1);
            }
            @Override
            protected void hookOnNext(Integer value) {
                System.out.println("next"+value);
                request(1);
                if (value == 3){
                    cancel(); //取消
                }
            }
            @Override
            protected void hookOnCancel() {
                System.out.println("cancel");
            }
            @Override
            protected void hookOnComplete() {
                System.out.println("complete");
            }
            @Override
            protected void hookOnError(Throwable throwable) {
                System.out.println("error"+throwable);

            }

            @Override
            protected void hookFinally(SignalType type) {
                System.out.println("最终回调"+type);
            }
        });
       // 声明周期

    }
}
